package com.shujia.sql

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object Demo9Jdbc {
  def main(args: Array[String]): Unit = {
    val setting: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()

    val table: TableEnvironment = TableEnvironment.create(setting)


    /**
      * 创建source表
      *
      */
    table.executeSql(
      """
        |CREATE TABLE students (
        |    id BIGINT,
        |    name STRING,
        |    age BIGINT,
        |    gender STRING,
        |    clazz STRING,
        |    PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'students',
        |   'username' ='root',
        |   'password' = '123456'
        |)
        |
      """.stripMargin)

    /**
      * jdbc sink表
      *
      * PRIMARY KEY:  一般以查询语句的分组条件作为主键
      *
      */

    table.executeSql(
      """
        |CREATE TABLE gender_num (
        |    gender STRING,
        |    num BIGINT,
        |    PRIMARY KEY (gender) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'gender_num',
        |   'username' ='root',
        |   'password' = '123456'
        |)
      """.stripMargin)

    table.executeSql(
      """
        |insert into gender_num
        |select gender,count(1) as num
        |from students
        |group by gender
        |
      """.stripMargin)
  }

}
